package com.lzwk.app.dwd.ocpx;

import com.lzwk.app.OcpxSum;
import com.lzwk.app.function.GetMixedOrderId;
import com.lzwk.base.BaseAppV1;
import com.lzwk.common.ocpx.OCPXConstant;
import com.lzwk.common.test.TestConstant;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.CoGroupFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * @title: TableTest
 * @Author Yeaser
 * @Date: 2022/4/13 15:24
 * @Version 1.0
 * Description: TODO
 */
public class TableTest extends BaseAppV1 {
    public static void main(String[] args) throws Exception {
        new TableTest().init(1,"cks","ocpx_tmp","test");
    }

    @Override
    protected void run(StreamExecutionEnvironment env, StreamTableEnvironment tenv) {
        tenv.executeSql("create table ocpx_account_wxuser(" +
                " `id` INT, " +
                " `openid` STRING, " +
                " `unionid` STRING, " +
                " `headimgurl` STRING, " +
                " `sex` INT, " +
                " `nickname` STRING, " +
                " `subscribed` INT, " +
                " `create_time` TIMESTAMP(0), " +
                " `active_ip` STRING, " +
                " `active_time` TIMESTAMP(0), " +
                " `update_time` TIMESTAMP(0), " +
                " `account_id` INT, " +
                " `wxapp_id` INT, " +
                " `proc_time` AS PROCTIME(), " +
                "  PRIMARY KEY (`id`) NOT ENFORCED" +
                " )with(" +
                "   'connector'='mysql-cdc', " +
                "   'hostname'='" + TestConstant.TESTHOST + "', " +
                "   'port'='" + TestConstant.TESTPORT + "', " +
                "   'username'='" + TestConstant.TESTUSERNAME + "', " +
                "   'password'='" + TestConstant.TESTPASSWORD + "', " +
                "   'database-name'='" + TestConstant.TESTOCPXDB + "', " +
                "   'table-name'='" + TestConstant.TESTOCPXJDBCACCOUNTWXUSER + "', " +
                "   'server-time-zone'='Asia/Shanghai', " +
                "   'debezium.snapshot.locking.mode' = 'none', " +
                "   'scan.incremental.snapshot.chunk.size' = '80000' " +
                ")");


        tenv.executeSql("create table mysql_pay_order(" +
                " `id` INT, " +
                " `payee_id` INT, " +
                " `payee_type` STRING, " +
                " `fee` INT, " +
                " `product_type` INT, " +
                " `product_id` INT, " +
                " `source_id` INT, " +
                " `skin_id` INT, " +
                " `wxapp_id` INT, " +
                " `paid` INT, " +
                " `deliver_status` INT, " +
                " `refund_status` INT, " +
                " `create_time` TIMESTAMP(0), " +
                " `pay_time` TIMESTAMP(0), " +
                " `callback_time` TIMESTAMP(0), " +
                " `outer_order_id` STRING, " +
                " `meta` STRING NOT NULL, " +
                " `ip` STRING, " +
                " `payer_id` INT, " +
                " `source` STRING, " +
                " `is_notified` INT, " +
                " `update_time` TIMESTAMP(0), " +
                " `openid` STRING, " +
                " `proc_time` AS PROCTIME(), " +
                "  PRIMARY KEY (`id`) NOT ENFORCED" +
                " )with(" +
                "   'connector'='mysql-cdc', " +
                "   'hostname'='" + TestConstant.TESTHOST + "', " +
                "   'port'='" + TestConstant.TESTPORT + "', " +
                "   'username'='" + TestConstant.TESTUSERNAME + "', " +
                "   'password'='" + TestConstant.TESTPASSWORD + "', " +
                "   'database-name'='" + TestConstant.TESTOCPXDB + "', " +
                "   'table-name'='" + TestConstant.TESTOCPXPAYORDER + "', " +
                "   'server-time-zone'='Asia/Shanghai', " +
                "   'server-id'='35081-35090', " +
                "   'debezium.snapshot.locking.mode' = 'none', " +
                "   'scan.incremental.snapshot.chunk.size' = '80000' " +
                ")");


        tenv.executeSql(" CREATE TABLE ods_ocpx_account_account ( " +
                " `id` INT, " +
                " `openid` STRING, " +
                " `unionid` STRING, " +
                " `nickname` STRING, " +
                " `headimgurl` STRING, " +
                " `sex` INT, " +
                " `province` STRING, " +
                " `country` STRING, " +
                " `city` STRING, " +
                " `status` INT, " +
                " `is_robot` INT, " +
                " `active_ip` STRING, " +
                " `active_time` TIMESTAMP(0), " +
                " `create_time` TIMESTAMP(0), " +
                " `is_init` INT, " +
                " `src` STRING, " +
                " `parent_account_id` INT, " +
                "  PRIMARY KEY (`id`) NOT ENFORCED" +
                ") with ( " +
                "  'connector' = 'jdbc', " +
                "  'driver' = 'com.mysql.cj.jdbc.Driver', " +
                "  'url'='jdbc:mysql://" + TestConstant.TESTHOST + ":" + TestConstant.TESTPORT + "/" + TestConstant.TESTOCPXDB + "?characterEncoding=utf8&serverTimezone=Asia/Shanghai&autoReconnect=true&tinyInt1isBit=false&transformedBitIsBoolean=false', " +
                "  'table-name' = " + "'" + TestConstant.TESTOCPXJDBCACCOUNTACCOUNT + "'" + ", " +
                "  'username'=" + "'" + TestConstant.TESTUSERNAME + "'" + ", " +
                "  'password'=" + "'" + TestConstant.TESTPASSWORD + "'" + ", " +
                "  'lookup.cache.max-rows'='2000', " +
                "  'lookup.cache.ttl'='600000' " +
                ")");


        tenv.executeSql(" CREATE TABLE ocpx_pay_order ( " +
                " `id` INT, " +
                " `payee_id` BIGINT, " +
                " `payee_type` STRING, " +
                " `fee` BIGINT, " +
                " `product_type` BIGINT, " +
                " `product_id` BIGINT, " +
                " `source_id` BIGINT, " +
                " `skin_id` BIGINT, " +
                " `wxapp_id` BIGINT, " +
                " `paid` INT, " +
                " `deliver_status` BIGINT, " +
                " `refund_status` BIGINT, " +
                " `create_time` TIMESTAMP(0), " +
                " `pay_time` TIMESTAMP(0), " +
                " `callback_time` TIMESTAMP(0), " +
                " `outer_order_id` STRING, " +
                " `meta` STRING NOT NULL, " +
                " `ip` STRING, " +
                " `payer_id` INT, " +
                " `source` STRING, " +
                " `is_notified` INT, " +
                " `update_time` TIMESTAMP(0), " +
                " `openid` STRING, " +
                " `proc_time` AS PROCTIME(), " +
                "  PRIMARY KEY (`id`) NOT ENFORCED" +
                ") with ( " +
                "  'connector' = 'jdbc', " +
                "  'driver' = 'com.mysql.cj.jdbc.Driver', " +
                "  'url'='jdbc:mysql://" + TestConstant.TESTHOST + ":" + TestConstant.TESTPORT + "/" + TestConstant.TESTOCPXDB + "?characterEncoding=utf8&serverTimezone=Asia/Shanghai&autoReconnect=true&tinyInt1isBit=false&transformedBitIsBoolean=false', " +
                "  'table-name' = " + "'" + TestConstant.TESTOCPXPAYORDER + "'" + ", " +
                "  'username'=" + "'" + TestConstant.TESTUSERNAME + "'" + ", " +
                "  'password'=" + "'" + TestConstant.TESTPASSWORD + "'" + ", " +
                "  'lookup.cache.max-rows'='2000', " +
                "  'lookup.cache.ttl'='600000' " +
                ")");


        tenv.executeSql("create table dwd_ocpx_pay_order(" +
                "   `id` INT, " +
                "   `create_time` TIMESTAMP(0), " +
                "   `pay_time` TIMESTAMP(0), " +
                "   `wxapp_id` INT, " +
                "   `payer_id` INT, " +
                "   `paid` INT, " +
                "   `account_id` INT, " +
                "   `wx_id` INT, " +
                "   `subscribed` INT, " +
                "   PRIMARY KEY (`id`, `payer_id`) NOT ENFORCED" +
                ")with(" +
                "  'connector' = 'jdbc', " +
                "  'driver' = 'com.mysql.cj.jdbc.Driver', " +
                "  'url'='jdbc:mysql://" + TestConstant.TESTHOST + ":" + TestConstant.TESTPORT + "/" + TestConstant.TESTOCPXDB + "?characterEncoding=utf8&serverTimezone=Asia/Shanghai&autoReconnect=true&tinyInt1isBit=false&transformedBitIsBoolean=false', " +
                "  'table-name' = " + "'" + TestConstant.TESTJOINTABLE + "'" + ", " +
                "  'username'=" + "'" + TestConstant.TESTUSERNAME + "'" + ", " +
                "  'password'=" + "'" + TestConstant.TESTPASSWORD + "'" + ", " +
                "  'lookup.cache.max-rows'='2000', " +
                "  'lookup.cache.ttl'='600000' " +
                ")");


        tenv.executeSql("CREATE TABLE ocpx_product_product (" +
                "  id INT, " +
                "  name string, " +
                "  tag_id INT, " +
                "  src_prod_id INT, " +
                "  src_prod_type string, " +
                "  primary key (id) not enforced " +
                ") with ( " +
                "  'connector'='jdbc', " +
                "  'url'='jdbc:mysql://" + OCPXConstant.OCPXSLAVEHOST + ":" + OCPXConstant.OCPXSLAVEPORT + "/" + OCPXConstant.OCPXSLAVEDBNAME + "?characterEncoding=utf8&serverTimezone=Asia/Shanghai&autoReconnect=true&tinyInt1isBit=false&transformedBitIsBoolean=false', " +
                "  'table-name' = " + "'" + OCPXConstant.OCPXJDBCPRODUCTPRODUCT + "'" + ", " +
                "  'username'=" + "'" + OCPXConstant.OCPXSLAVEUSERNAME + "'" + ", " +
                "  'password'=" + "'" + OCPXConstant.OCPXSLAVEPASSWORD + "'" + ", " +
                "  'lookup.cache.max-rows'='10', " +
                "  'lookup.cache.ttl'='600000' " +
                ")");


        Table wxuserTable = tenv.sqlQuery(" select * from ocpx_account_wxuser ");

        Table orderTable = tenv.sqlQuery("select t1.id," +
                " t1.create_time, " +
                " t1.pay_time, " +
                " t1.wxapp_id, " +
                " t1.payer_id, " +
                " t1.paid, " +
                " ifnull(`t2`.`parent_account_id`, `t2`.`id`) as account_id " +
                " from mysql_pay_order t1 " +
                " left join ods_ocpx_account_account FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t2 " +
                " on t1.payer_id = t2.id " +
                "");

        tenv.createTemporaryFunction("GetMixedOrderId", GetMixedOrderId.class);

        Table resTable = tenv.sqlQuery(
                " select" +
                        " cast(t3.id as string) id1,  " +
                        " cast(t4.id as string) id2,  " +
                        " if(cast(t3.id as string) is null,if(cast(t4.id as string) is null,'',cast(t4.id as string)),if(cast(t4.id as string) is null,cast(t3.id as string),CONCAT_WS(',',cast(t3.id as string),cast(t4.id as string)))) as ids,  " +
                        " t1.id as wx_id, " +
                        " order_id, " +
                        " t1.wxapp_id as wxapp_id, " +
                        " t1.account_id as account_id, " +
                        " t1.subscribed as subscribed " +
                        " from ocpx_account_wxuser t1 " +
                        " left join ods_ocpx_account_account FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t2 " +
                        "   on t1.account_id = t2.parent_account_id  " +
                        " left join ocpx_pay_order FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t3 " +
                        "   on cast(t1.wxapp_id as bigint) = t3.wxapp_id  " +
                        "   and t2.id = t3.payer_id " +
                        " left join ocpx_pay_order FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t4 " +
                        "   on cast(t1.wxapp_id as bigint) = t4.wxapp_id  " +
                        "   and t1.account_id = t4.payer_id " +
                        " left join ocpx_product_product FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t5 " +
                        "   on cast(ifnull(t3.product_id,t4.product_id) as int)= t5.id " +
                        " join lateral table(GetMixedOrderId(if(cast(t3.id as string) is null,if(cast(t4.id as string) is null,'',cast(t4.id as string)),if(cast(t4.id as string) is null,cast(t3.id as string),CONCAT_WS(',',cast(t3.id as string),cast(t4.id as string)))))) as t (order_id) on true " +
                        " where t5.src_prod_type = 'wk_act' "
        );

        resTable.execute().print();



        tenv.createTemporaryView("resTable",resTable);



        tenv.sqlQuery(" select id,id1,id2,wx_id,wxapp_id,account_id,subscribed " +
                " from resTable" +
                " join lateral table(GetMixedOrderId(ids)) as t (id) on true ");




        /*SingleOutputStreamOperator<TestOrder> orderDS = tenv.toRetractStream(orderTable, TestOrder.class)
                .filter(data -> data.f0)
                .map(data -> data.f1)
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<TestOrder>forBoundedOutOfOrderness(Duration.ofMinutes(2))
                        .withTimestampAssigner((data, ts) -> data.getCreate_time().getTime()));

        //orderDS.print("=======");

        SingleOutputStreamOperator<TestWxuser> wxuserDS = tenv.toRetractStream(wxuserTable, TestWxuser.class)
                .filter(data -> data.f0)
                .map(data -> data.f1)
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<TestWxuser>forBoundedOutOfOrderness(Duration.ofMinutes(2))
                        .withTimestampAssigner((data, ts) -> data.getCreate_time().getTime()));


        DataStream<TestBean> resDS = orderDS.coGroup(wxuserDS)
                .where(new KeySelector<TestOrder, Tuple2<Integer, Integer>>() {
                    @Override
                    public Tuple2<Integer, Integer> getKey(TestOrder value) throws Exception {
                        return new Tuple2<>(value.getAccount_id(), value.getWxapp_id());
                    }
                })
                .equalTo(new KeySelector<TestWxuser, Tuple2<Integer, Integer>>() {
                    @Override
                    public Tuple2<Integer, Integer> getKey(TestWxuser value) throws Exception {
                        return new Tuple2<>(value.getAccount_id(), value.getWxapp_id());
                    }
                })
                .window(TumblingProcessingTimeWindows.of(Time.seconds(10)))
                .apply(new CoGroupFunction<TestOrder, TestWxuser, TestBean>() {
                    @Override
                    public void coGroup(Iterable<TestOrder> first, Iterable<TestWxuser> second, Collector<TestBean> out) throws Exception {
                        for (TestOrder testOrder : first) {
                            boolean flag = false;
                            for (TestWxuser testWxuser : second) {
                                out.collect(new TestBean(testOrder, testWxuser));
                                flag = true;
                            }
                            if (!flag) {
                                out.collect(new TestBean(testOrder));
                            }
                        }
                        for (TestWxuser testWxuser : second) {
                            for (TestOrder testOrder : first) {
                                out.collect(new TestBean(testOrder, testWxuser));
                            }
                        }
                    }
                });

        tenv.createTemporaryView("restable",resDS);

        tenv.executeSql(" insert into dwd_ocpx_pay_order select * from restable  ");*/


/*        tenv.executeSql(" insert into  dwd_ocpx_pay_order " +
                " select " +
                " tm.id, " +
                " tm.create_time, " +
                " tm.pay_time, " +
                " tm.wxapp_id,  " +
                " tm.payer_id, " +
                " tm.paid, " +
                " tm.account_id, " +
                " t3.id as wx_id, " +
                " t3.subscribed " +
                " from " +
                "(select t1.id," +
                "   t1.create_time, " +
                "   t1.pay_time, " +
                " t1.wxapp_id, " +
                " t1.payer_id, " +
                " t1.paid, " +
                " ifnull(`t2`.`parent_account_id`, `t2`.`id`) as account_id " +
                " from mysql_pay_order t1 " +
                " left join ods_ocpx_account_account FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t2 " +
                " on t1.payer_id = t2.id " +
                " ) tm ,  ocpx_account_wxuser t3 " +
                " where 1=1 " +
                " and tm.account_id = t3.account_id " +
                " and tm.wxapp_id = t3.wxapp_id " +
                " and tm.create_time between t3.update_time - interval '3' year and t3.update_time +  interval '3' year "+
                " and date_format(tm.create_time,'yyyy-MM-dd') = '2022-04-19' " +
                " and date_format(t3.create_time,'yyyy-MM-dd') = '2022-04-19' "
        );*/


        /*tenv.executeSql(" insert into  dwd_ocpx_pay_order " +
                " select " +
                " tm.id, " +
                " tm.create_time, " +
                " tm.pay_time, " +
                " tm.wxapp_id,  " +
                " tm.payer_id, " +
                " tm.paid, " +
                " tm.account_id, " +
                " t3.id as wx_id, " +
                " t3.subscribed " +
                " from " +
                "(select t1.id," +
                "   t1.create_time, " +
                "   t1.pay_time, " +
                " t1.wxapp_id, " +
                " t1.payer_id, " +
                " t1.paid, " +
                " ifnull(`t2`.`parent_account_id`, `t2`.`id`) as account_id " +
                " from mysql_pay_order t1 " +
                " left join ods_ocpx_account_account FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t2 " +
                " on t1.payer_id = t2.id " +
                " ) tm left join  ocpx_account_wxuser t3 " +
                " on tm.account_id = t3.account_id " +
                " and tm.wxapp_id = t3.wxapp_id " +
                " and tm.create_time between t3.update_time - interval '3' year and t3.update_time +  interval '3' year "+
                " where date_format(tm.create_time,'yyyy-MM-dd') = '2022-04-19' " +
                " and date_format(t3.create_time,'yyyy-MM-dd') = '2022-04-19' "
        );*/

    }
}
